package com.uestc.coprocessor;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;

import com.google.protobuf.ServiceException;
import com.uestc.util.JsonUtil;
import com.uestc.util.Log;

public class MultiColumnAggregateClient {

	public static class CountInfo {
		public long qiniuCount;
		public long qiniu2Count;
		public long jinshanCount;
		public long otherCount;

		@Override
		public String toString() {
			String rString = String.format("qiniuCount:%d;Qiniu2Count:%d;JinshanCount:%d;OtherCount:%d",
					this.qiniuCount, this.qiniu2Count, this.jinshanCount, this.otherCount);
			return rString;
		}

		/**
		 * 生成jsonString
		 * 
		 * @return
		 * @throws IllegalArgumentException
		 * @throws IllegalAccessException
		 */
		public String toJsonString() throws IllegalArgumentException, IllegalAccessException {
			Field[] fields = this.getClass().getFields();
			Map<String, Object> map = new LinkedHashMap<>();
			for (Field field : fields) {
				map.put(field.getName(), field.get(this));
			}
			return JsonUtil.createJsonString(map);
		}
	}

	public CountInfo getSeveralCounts(Table table, String columnNames) throws ServiceException, Throwable {
		final JAggregateProtocol.CountRequest request = JAggregateProtocol.CountRequest.newBuilder()
				.setColumns(columnNames).build();
		Map<byte[], CountInfo> map = table.coprocessorService(JAggregateProtocol.RowCountService.class, null, null,
				new Batch.Call<JAggregateProtocol.RowCountService, CountInfo>() {

					@Override
					public CountInfo call(JAggregateProtocol.RowCountService aggregate) throws IOException {
						BlockingRpcCallback<JAggregateProtocol.CountResponse> rpcCallback = new BlockingRpcCallback<>();
						aggregate.getSizes(null, request, rpcCallback);
						JAggregateProtocol.CountResponse response = rpcCallback.get();
						CountInfo countInfo = new CountInfo();
						countInfo.qiniuCount = response.getQiniuCount();
						countInfo.qiniu2Count = response.getQiniu2Count();
						countInfo.jinshanCount = response.getJinshanCount();
						countInfo.otherCount = response.getOtherCloud();
						return countInfo;
					}
				});
		CountInfo result = new CountInfo();
		for (CountInfo countInfo : map.values()) {
			result.qiniuCount += countInfo.qiniuCount;
			result.qiniu2Count += countInfo.qiniu2Count;
			result.jinshanCount += countInfo.jinshanCount;
			result.otherCount += countInfo.otherCount;
		}
		Log.logger.info(result);
		return result;
	}
}
